package belf.migrate.engine.verify;

import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.exception.TableNotExistException;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.model.VerifyModel;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.ColumnMetaData;
import belf.migrate.api.table.schema.TablePath;
import belf.migrate.api.taskconf.SinkConf;
import belf.migrate.api.taskconf.SourceConf;
import belf.migrate.api.util.DatabaseType;
import belf.migrate.api.util.FieldCase;
import belf.migrate.core.jdbc.DBTypeMapping;
import belf.migrate.core.jdbc.TypeModel;
import belf.migrate.engine.job.CatalogCreator;
import belf.migrate.engine.job.Result;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
public class VerifyTable {

    private static Gson gson = new Gson();
    public static JsonObject verify(VerifyModel verifyModel) {

        SourceConf sourceConf = verifyModel.getSourceConf();
        SinkConf sinkConf = verifyModel.getSinkConf();
        try {
            AbstractJdbcCatalog sourceCatalog = CatalogCreator.createCatalog(
                    sourceConf.getCatalogName(),
                    sourceConf.getDatabaseName(),
                    sourceConf.getSchemaName(), sourceConf.getConnectionConf(), sourceConf.isCaseSensitive());
            AbstractJdbcCatalog sinkCatalog = CatalogCreator.createCatalog(
                    sinkConf.getCatalogName(),
                    sinkConf.getDatabaseName(),
                    sinkConf.getSchemaName(), sinkConf.getConnectionConf(), sinkConf.isCaseSensitive());

            DatabaseType sourceDatabaseType = DatabaseType.getDatabaseType(sourceConf.getCatalogName());
            DatabaseType sinkDatabaseType = DatabaseType.getDatabaseType(sinkConf.getCatalogName());

            QualifiersModel sourceQualifiersModel = new QualifiersModel("", "", FieldCase.getFieldCase(sourceConf.getFieldCase()), sourceDatabaseType.getQualifiedType());
            QualifiersModel sinkQualifiersModel = new QualifiersModel("", "", FieldCase.getFieldCase(sinkConf.getFieldCase()), sinkDatabaseType.getQualifiedType());

            TablePath sourceTablePath = TablePath.of(
                    sourceConf.getDatabaseName(),
                    sourceConf.getSchemaName(),
                    verifyModel.getName(),
                    sourceQualifiersModel);
            TablePath sinkTablePath = TablePath.of(
                    sinkConf.getDatabaseName(),
                    sinkConf.getSchemaName(),
                    verifyModel.getName(),
                    sinkQualifiersModel);

            String taskType = sourceConf.getCatalogName() + "-" + sinkConf.getCatalogName();
            Map<String, TypeModel> mapping = DBTypeMapping.getInstance().getMapping(
                    taskType);

            if (!sinkCatalog.tableExists(sinkTablePath)) {
                return gson.toJsonTree(Result.fail("表未创建成功，表不存在！")).getAsJsonObject();
            }
            List<ColumnMetaData> sourceColumns = sourceCatalog.getAllColumnsBySql(sourceTablePath);
            List<ColumnMetaData> sinkColumns = sinkCatalog.getAllColumnsBySql(sinkTablePath);
            if (sinkColumns == null || sourceColumns.size() != sinkColumns.size()) {
                return gson.toJsonTree(Result.fail("表未创建成功，表字段数量不一致！")).getAsJsonObject();
            }
            Map<String, ColumnMetaData> sinkDataMap = sinkColumns.stream().collect(
                    Collectors.toMap(ColumnMetaData -> ColumnMetaData.getColumnName().toLowerCase(Locale.ROOT),
                            ColumnData -> ColumnData));
            for (ColumnMetaData sourceColumn : sourceColumns) {
                ColumnMetaData sinkColumn = sinkDataMap.get(sourceColumn.getColumnName().toLowerCase(Locale.ROOT));
                if (sinkColumn == null) {
                    return gson.toJsonTree(Result.fail("表未创建成功，表字段不一致！")).getAsJsonObject();
                }
                if (!sinkColumn.getColumnTypeName().equalsIgnoreCase(mapping.get(sourceColumn.getColumnTypeName().toUpperCase(Locale.ROOT)).getTo())) {
                    return gson.toJsonTree(Result.fail("表未创建成功，表字段类型不一致！")).getAsJsonObject();
                }
            }

        } catch (CatalogException | TableNotExistException e) {
            log.error("校验表失败", e);
            return gson.toJsonTree(Result.fail(e.getMessage())).getAsJsonObject();
        }
        return gson.toJsonTree(Result.success()).getAsJsonObject();

    }
}
